package com.zyx.flinkdemo.stream.source;

import com.zyx.flinkdemo.pojo.WordCountTs;
import com.zyx.flinkdemo.stream.utils.TimeUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zyx
 * @since 2021/5/27 23:33
 * desc:
 */
public class SourceForWindow implements SourceFunction<WordCountTs>  {

    public static Logger LOG = LoggerFactory.getLogger(SourceForWindow.class);

    private static final long serialVersionUID = 1L;

    public static final String[] WORDS = new String[]{
            "zyx", "zyx", "zyx", "zyx", "zyx",
            "java", "java",
            "flink", "flink", "flink",
            "hadoop", "hadoop",
            "spark", "spark"
    };

    private volatile boolean isRunning = true;

    private final long sleepTime;

    private Boolean stopSession = false;



    public SourceForWindow(long sleepTime, Boolean stopSession) {
        this.sleepTime = sleepTime;
        this.stopSession = stopSession;
    }

    public SourceForWindow(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    @Override
    public void run(SourceFunction.SourceContext<WordCountTs> ctx) throws Exception {
        int count = 0;
        while (isRunning) {
            String word = WORDS[count % WORDS.length];
            long ts = System.currentTimeMillis();
            String time = TimeUtils.getHhmmss(ts);
            WordCountTs wordCountTs = new WordCountTs(word, count, time, ts);
            ctx.collect(wordCountTs);
            // LOG.info("send data :" + tuple2);
            // System.out.println("send data :" + tuple2);
            if (stopSession && count % WORDS.length == 0) {
                Thread.sleep(10000);
            } else {
                Thread.sleep(sleepTime);
            }
            count++;
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

}
